package com.pony.test;

import java.io.File;
import java.io.IOException;
import java.util.Iterator;

import net.paoding.analysis.analyzer.PaodingAnalyzer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;

public class IndexReducer extends MapReduceBase implements
		Reducer<IntWritable, IntermediateForm, Text, Text> {

	@Override
	public void reduce(IntWritable inputKey, Iterator<IntermediateForm> inputValues,
			OutputCollector<Text, Text> output, Reporter arg3) throws IOException {
		
		Analyzer analyzer = new PaodingAnalyzer();
		
		IndexWriter writer = new IndexWriter(FSDirectory.open(new File("e:/indexPaoding/shards/0000")),
                analyzer,true,IndexWriter.MaxFieldLength.UNLIMITED);
		
		while(inputValues.hasNext()){
			IntermediateForm form = inputValues.next();
			form.closeWriter();
			writer.addIndexesNoOptimize(new Directory[]{form.getDirectory()});
			System.out.println("***************");
		}
		writer.optimize();
		writer.close();
		
		output.collect(new Text("index"), new Text("done"));
		
		System.out.println("--------------DONE---------------------");
	}

}
